library(RMySQL)
library(multicore)

dbconnect <- function() {

  con = dbConnect(MySQL(), user="cgp", password="duckymomo", dbname="smdata", host="ec2-23-20-125-145.compute-1.amazonaws.com")
  return(con)
}

createcorrtable <- function(con) {
  if (dbExistsTable(con, "correlations")) {
    dbRemoveTable(con, "correlations")
  }
  statement = "CREATE TABLE correlations (symbolA VARCHAR(40), symbolB VARCHAR(40), corr FLOAT, PRIMARY KEY(symbolA, symbolB));"
  dbSendQuery(con, statement)
}

getsymbols <- function(con, exchange) {
  statement = paste("select symbol from symbols where exchange=", "\"", exchange, "\";", sep="")
  
  sym = dbGetQuery(con, statement)
  return(sym)
}

getdata <- function(con, symbol) {
  statement = paste("select name,last_price,last_trade_date,last_trade_time from quotes where symbol=", "\"", symbol, "\";", sep="")
  data = dbGetQuery(con, statement)
  
  numrows = nrow(data)
  if (numrows == 0) {
  	return(NULL)
  }
  
  numcols = 2
  symboldata = matrix(1:numrows*numcols, nrow=numrows, ncol=numcols)
  for (row in c(1:numrows)) {
    thisdate = data$last_trade_date[row]
    thistime = data$last_trade_time[row]
    dtstr = paste(thisdate, thistime)
    res = try(as.numeric(as.POSIXct(dtstr)))
    if (class(res) == "try-error") {
      next
    }
    symboldata[row,1] = res
    res = try(as.numeric(data$last_price[row]))
    if (class(res) == "try-error") {
      next
    }
    symboldata[row,2] = res
  } 
  
  return(symboldata)
}

interpdata <- function(data, mintime, maxtime, interval) {
  timestep = seq(mintime, maxtime, by=interval)
  interpprice = approx(data[,1], data[,2], timestep)
  return(c(timestep, interpprice))
}

calccorr <- function(symA, symB) {
  con = dbconnect()
  dataA = getdata(con, symA)
  if (is.null(dataA) == TRUE) {
  	print(paste("Warning: Data for symbol", symA, "is NULL."))
  	dbclose(con)
  	return(-999)
  }
  dataB = getdata(con, symB)
  if (is.null(dataB) == TRUE) {
  	print(paste("Warning: Data for symbol", symB, "is NULL."))
  	dbclose(con)
  	return(-999)
  }
  dbclose(con)
  
  mintimeA = min(dataA[,1])
  maxtimeA = max(dataA[,1])
  mintimeB = min(dataB[,1])
  maxtimeB = max(dataB[,1])
  start = max(mintimeA, mintimeB)
  end = min(maxtimeA, maxtimeB)
  interval = 40000
  if (end < start) {
    print(paste("error, end time less than start time for ", symA, symB, "no time overlap"))
    return(-999)
  }
  timestep = seq(start, end, by=interval)
  dataAinterp = approx(dataA[,1], dataA[,2], timestep)
  dataBinterp = approx(dataB[,1], dataB[,2], timestep)
  res = try(cor(dataAinterp$y, dataBinterp$y))
  if (class(res) == "try-error") {
    return(-999)
  }
  return(res)
}




runexchange <- function(exchange) {
  con = dbconnect()
  sym = getsymbols(con, exchange)
  len = length(sym$symbol)

  for (ii in c(1:len)) {
    symbolA = sym$symbol[ii]

    for (jj in c(ii:len)) {  
      if (jj == ii) {
        next
      }
      symbolB = sym$symbol[jj]
      cc = calccorr(con, symbolA, symbolB)
      if (is.na(cc)) {
        cc = -999.0
      }
      print(paste(symbolA, ":", symbolB, ":", cc))
      statement = paste("REPLACE into correlations VALUES (", "\"", symbolA, "\",", "\"", symbolB, "\",", cc, ");", sep="")
      dbSendQuery(con, statement)
      statement = paste("REPLACE into correlations VALUES (", "\"", symbolB, "\",", "\"", symbolA, "\",", cc, ");", sep="")
      dbSendQuery(con, statement)
    }

  }
  dbclose()

}

tf <- function(x, y) {
	return(x+y)
}

runexchangeparallel <- function(exchange) {
  con = dbconnect()
  sym = getsymbols(con, exchange)
  dbclose(con)
  
  sym = unlist(sym)
  for (ii in c(1:length(sym))) {
    symbolA = sym[ii]
    if (ii < length(sym)) {
	    remsymbols = sym[ii:length(sym)]
	    print(paste("Processing symbol", symbolA, "..."))
	    corrres = mclapply(remsymbols, calccorr, symbolA)
	    badidx = which(is.na(corrres))
	    if (length(badidx) != 0) {
	      corrres[badidx] = -999.0
	    }
	    filename = paste("~/", symbolA,".txt", sep="")
	    print(paste("Writing results for", symbolA, "to", filename))
	    write(rbind(remsymbols,unlist(corrres)), file=filename, sep=",", ncolumns=2)
    }
  }
  
}



dbclose <- function(con, drv) {
  dbDisconnect(con)
}
